package com.linkstec.mot.service;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.linkstec.mot.queue.DataQueue;
import com.linkstec.mot.serializer.IParser;
import com.linkstec.mot.util.ConfigUtil;
import com.linkstec.mot.util.ExceptionTracker;

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

/**
 * KAFKA消息接收类
 * 
 * @author PENGYAO
 * 
 */
public class MsgParserService implements Runnable {

	private static Logger logger = LoggerFactory
			.getLogger(MsgParserService.class);

	private KafkaStream<byte[], byte[]> kafkaStream;

	public MsgParserService(KafkaStream<byte[], byte[]> kafkaStream) {
		this.kafkaStream = kafkaStream;
	}

	/**
	 * WSLogDto所需属性为外部传入
	 * 
	 * @param
	 */
	public void run() {
		if (null == parser) {
			logger.error(String.format("Excetion>>>>>>>>>%s",
					"kafka message parser is null...kafka consumer exit..."));
			return;
		}
		ConsumerIterator<byte[], byte[]> it = kafkaStream.iterator();

		while (true) {
			try {
				if (it.hasNext()) {
					byte[] receiveStr = it.next().message();
					logger.info("receive a message from kafka...");
					Object data = parser.parse(receiveStr);
					DataQueue.put(data);
				}
			} catch (Exception e) {
				logger.error(ExceptionTracker.trace(e));
			}
		}
	}

	private static IParser parser = null;

	static {
		// 实例化消息解析器
		String parserClassPath = ConfigUtil.getProperty("parser",
				"com.linkstec.mot.serializer.StringParser");
		try {
			Class<?> clazz = Class.forName(parserClassPath);

			Constructor<?> constructor = clazz.getConstructor(String.class);

			parser = (IParser) constructor.newInstance("UTF8");

			logger.info("kafka message parser is:" + parserClassPath);

		} catch (ClassNotFoundException e) {
			logger.error(ExceptionTracker.trace(e));
		} catch (InstantiationException e) {
			logger.error(ExceptionTracker.trace(e));
		} catch (IllegalAccessException e) {
			logger.error(ExceptionTracker.trace(e));
		} catch (NoSuchMethodException e) {
			logger.error(ExceptionTracker.trace(e));
		} catch (SecurityException e) {
			logger.error(ExceptionTracker.trace(e));
		} catch (IllegalArgumentException e) {
			logger.error(ExceptionTracker.trace(e));
		} catch (InvocationTargetException e) {
			logger.error(ExceptionTracker.trace(e));
		}

	}

}
